/*-------------------------------------------------------------------------
 *
 *	parallel_slot.c
 *		前端并行数据库连接的并行支持
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * src/fe_utils/parallel_slot.c
 *
 *-------------------------------------------------------------------------
 */

#ifdef WIN32
#define FD_SETSIZE 1024			/* 必须在包含 winsock2.h 之前设置 */
#endif

#include "postgres_fe.h"

#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif

#include "common/logging.h"
#include "fe_utils/cancel.h"
#include "fe_utils/parallel_slot.h"
#include "fe_utils/query_utils.h"

#define ERRCODE_UNDEFINED_TABLE  "42P01"

static int	fc_select_loop(int fc_maxFd, fd_set *fc_workerset);
static bool fc_processQueryResult(ParallelSlot *fc_slot, PGresult *fc_result);

/*
 * 处理（并删除）查询结果。如果没有问题返回真，
 * 否则返回假。由处理程序决定什么构成一个
 * 问题。
 */
static bool fc_processQueryResult(ParallelSlot *fc_slot, PGresult *fc_result)
{
	Assert(fc_slot->handler != NULL);

	/* 失败时，处理程序应在释放结果后返回 NULL */
	if (!fc_slot->handler(fc_result, fc_slot->connection, fc_slot->handler_context))
		return false;

	/* 好吧，我们必须自己释放它 */
	PQclear(fc_result);
	return true;
}

/*
 * 消耗为给定连接生成的所有结果，直到
 * 没有什么剩余。如果遇到至少一个错误，返回假。
 * 请注意，如果连接忙，此操作将被阻塞。
 */
static bool fc_consumeQueryResult(ParallelSlot *fc_slot)
{
	bool		fc_ok = true;
	PGresult   *fc_result;

	SetCancelConn(fc_slot->connection);
	while ((fc_result = PQgetResult(fc_slot->connection)) != NULL)
	{
		if (!fc_processQueryResult(fc_slot, fc_result))
			fc_ok = false;
	}
	ResetCancelConn();
	return fc_ok;
}

/*
 * 等待给定集合中的文件描述符变为可读。
 *
 * 返回准备就绪的描述符数量，失败时返回 -1（包括
 * 获取取消请求）。
 */
static int fc_select_loop(int fc_maxFd, fd_set *fc_workerset)
{
	int			fc_i;
	fd_set		fc_saveSet = *fc_workerset;

	if (CancelRequested)
		return -1;

	for (;;)
	{
		/*
		 * 在 Windows 上，我们需要不时检查取消请求；
		 * 在其他平台上，我们依赖 select() 在被打断时返回。
		 */
		struct timeval *fc_tvp;
#ifdef WIN32
		struct timeval tv = {0, 1000000};

		tvp = &tv;
#else
		fc_tvp = NULL;
#endif

		*fc_workerset = fc_saveSet;
		fc_i = select(fc_maxFd + 1, fc_workerset, NULL, NULL, fc_tvp);

#ifdef WIN32
		if (i == SOCKET_ERROR)
		{
			i = -1;

			if (WSAGetLastError() == WSAEINTR)
				errno = EINTR;
		}
#endif

		if (fc_i < 0 && errno == EINTR)
			continue;			/* 忽略这个 */
		if (fc_i < 0 || CancelRequested)
			return -1;			/* 但不要忽略这个 */
		if (fc_i == 0)
			continue;			/* 超时（仅限 Win32） */
		break;
	}

	return fc_i;
}

/*
 * 返回合适的空闲槽的偏移量，如果没有可用的则返回 -1。如果
 * 给定的 dbname 不为 null，则仅考虑连接到给定
 * 数据库的空闲槽为合适，否则所有连接的空闲槽
 * 均视为合适。
 */
static int fc_find_matching_idle_slot(const ParallelSlotArray *fc_sa, const char *fc_dbname)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
	{
		if (fc_sa->slots[fc_i].inUse)
			continue;

		if (fc_sa->slots[fc_i].connection == NULL)
			continue;

		if (fc_dbname == NULL ||
			strcmp(PQdb(fc_sa->slots[fc_i].connection), fc_dbname) == 0)
			return fc_i;
	}
	return -1;
}

/*
 * 返回第一个没有数据库连接的槽的偏移量，如果
 * 所有槽都已连接，则返回 -1。
 */
static int fc_find_unconnected_slot(const ParallelSlotArray *fc_sa)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
	{
		if (fc_sa->slots[fc_i].inUse)
			continue;

		if (fc_sa->slots[fc_i].connection == NULL)
			return fc_i;
	}

	return -1;
}

/*
 * 返回第一个空闲槽的偏移量，如果所有槽都忙，则返回 -1。
 */
static int fc_find_any_idle_slot(const ParallelSlotArray *fc_sa)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
		if (!fc_sa->slots[fc_i].inUse)
			return fc_i;

	return -1;
}

/*
 * 等待任何槽的连接有查询结果，消耗结果，
 * 并根据需要更新槽的状态。成功时返回真，
 * 取消、错误或没有槽连接时返回假。
 */
static bool fc_wait_on_slots(ParallelSlotArray *fc_sa)
{
	int			fc_i;
	fd_set		fc_slotset;
	int			fc_maxFd = 0;
	PGconn	   *fc_cancelconn = NULL;

	/* 我们必须为每次调用 select_loop 重新构建 fd_set */
	FD_ZERO(&fc_slotset);

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
	{
		int			fc_sock;

		/* 如果仍然有没有连接的槽，我们不应该到这里 */
		Assert(fc_sa->slots[fc_i].connection != NULL);

		fc_sock = PQsocket(fc_sa->slots[fc_i].connection);

		/*
		 * 我们并不真正期望在启动后任何连接会失去其套接字，
		 * 但以防万一，可以通过忽略它们来应对。
		 */
		if (fc_sock < 0)
			continue;

		/* 跟踪我们看到的第一个有效连接。 */
		if (fc_cancelconn == NULL)
			fc_cancelconn = fc_sa->slots[fc_i].connection;

		FD_SET(fc_sock, &fc_slotset);
		if (fc_sock > fc_maxFd)
			fc_maxFd = fc_sock;
	}

	/*
	 * 如果我们到这里而没有有效连接，处理将无法
	 * 继续。
	 */
	if (fc_cancelconn == NULL)
		return false;

	SetCancelConn(fc_cancelconn);
	fc_i = fc_select_loop(fc_maxFd, &fc_slotset);
	ResetCancelConn();

	/* 失败？ */
	if (fc_i < 0)
		return false;

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
	{
		int			fc_sock;

		fc_sock = PQsocket(fc_sa->slots[fc_i].connection);

		if (fc_sock >= 0 && FD_ISSET(fc_sock, &fc_slotset))
		{
			/* select() 说输入可用，所以消耗它 */
			PQconsumeInput(fc_sa->slots[fc_i].connection);
		}

		/* 收集结果，只要有可用的结果 */
		while (!PQisBusy(fc_sa->slots[fc_i].connection))
		{
			PGresult   *fc_result = PQgetResult(fc_sa->slots[fc_i].connection);

			if (fc_result != NULL)
			{
				/* 处理并丢弃命令结果 */
				if (!fc_processQueryResult(&fc_sa->slots[fc_i], fc_result))
					return false;
			}
			else
			{
				/* 此连接已变为空闲 */
				fc_sa->slots[fc_i].inUse = false;
				ParallelSlotClearHandler(&fc_sa->slots[fc_i]);
				break;
			}
		}
	}
	return true;
}

/*
 * 使用存储的连接参数打开一个新的数据库连接，
 * 如果 dbname 不为 null，则可选地使用给定的 dbname，
 * 执行存储的初始命令（如果有），并将新连接与
 * 给定槽关联。
 */
static void fc_connect_slot(ParallelSlotArray *fc_sa, int fc_slotno, const char *fc_dbname)
{
	const char *fc_old_override;
	ParallelSlot *fc_slot = &fc_sa->slots[fc_slotno];

	fc_old_override = fc_sa->cparams->override_dbname;
	if (fc_dbname)
		fc_sa->cparams->override_dbname = fc_dbname;
	fc_slot->connection = connectDatabase(fc_sa->cparams, fc_sa->progname, fc_sa->echo, false, true);
	fc_sa->cparams->override_dbname = fc_old_override;

	/*
	 * POSIX定义FD_SETSIZE为FD_SET()及相关宏可接受的最高文件描述符。Windows将其定义为集合中文件描述符数量的上限，而不是每个文件描述符值的上限；请参见
	 * https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-select
	 * 和
	 * https://learn.microsoft.com/en-us/windows/win32/api/winsock/ns-winsock-fd_set。
	 * 我们不能忽视这一点，因为Windows的文件描述符起始值较高，重用延迟，并且跳过一些值。在十个以下的并发文件描述符快速开启和关闭的情况下，可以达到文件描述符1024。
	 *
	 * 在这里进行硬退出有点难看，但似乎没有必要为了让API更好看而复杂化处理。
	 */
#ifdef WIN32
	if (slotno >= FD_SETSIZE)
	{
		pg_log_error("too many jobs for this platform: %d", slotno);
		exit(1);
	}
#else
	{
		int			fc_fd = PQsocket(fc_slot->connection);

		if (fc_fd >= FD_SETSIZE)
		{
			pg_log_error("socket file descriptor out of range for select(): %d",
						 fc_fd);
			pg_log_error_hint("Try fewer jobs.");
			exit(1);
		}
	}
#endif

	/* 使用提供的命令设置连接（如果有）。 */
	if (fc_sa->initcmd)
		executeCommand(fc_slot->connection, fc_sa->initcmd, fc_sa->echo);
}

/*
 * ParallelSlotsGetIdle
 *		返回一个已准备好执行命令的连接槽。
 *
 * 返回的槽选择如下：
 *
 * 如果任何空闲槽已经有一个打开的连接，并且如果dbname为
 * null或现有连接指向给定数据库，则该槽将被
 * 返回，允许重用该连接。
 *
 * 否则，如果任何空闲槽尚未连接到任何数据库，则该槽
 * 将被返回，其连接使用存储的cparams打开，并且
 * 如果dbname不为null，则可以选择使用给定的dbname。
 *
 * 否则，如果存在任何空闲槽，将选择一个空闲槽，并在
 * 断开并重新连接到已存储的cparams后返回该槽，
 * 如果dbname不为null，则可以选择使用给定的dbname。
 *
 * 否则，如果任何槽具有忙碌的连接，我们将循环调用select()
 * 直到一个套接字变得可用。当发生这种情况时，我们读取整个
 * 集合并标记为可用的所有套接字为自由。然后我们使用
 * 上述相同的规则选择一个槽。
 *
 * 否则，我们无法返回槽，这就是错误，返回NULL。
 *
 * 对于任何创建的连接，如果存储的initcmd不为null，它将在返回槽之前
 * 作为命令在新形成的连接上执行。
 *
 * 如果发生错误，则返回NULL。
 */
ParallelSlot *
ParallelSlotsGetIdle(ParallelSlotArray *fc_sa, const char *fc_dbname)
{
	int			fc_offset;

	Assert(fc_sa);
	Assert(fc_sa->numslots > 0);

	while (1)
	{
		/* 第一选择：已连接到所需数据库的槽。 */
		fc_offset = fc_find_matching_idle_slot(fc_sa, fc_dbname);
		if (fc_offset >= 0)
		{
			fc_sa->slots[fc_offset].inUse = true;
			return &fc_sa->slots[fc_offset];
		}

		/* 第二选择：未连接到任何数据库的槽。 */
		fc_offset = fc_find_unconnected_slot(fc_sa);
		if (fc_offset >= 0)
		{
			fc_connect_slot(fc_sa, fc_offset, fc_dbname);
			fc_sa->slots[fc_offset].inUse = true;
			return &fc_sa->slots[fc_offset];
		}

		/* 第三选择：连接到错误数据库的槽。 */
		fc_offset = fc_find_any_idle_slot(fc_sa);
		if (fc_offset >= 0)
		{
			disconnectDatabase(fc_sa->slots[fc_offset].connection);
			fc_sa->slots[fc_offset].connection = NULL;
			fc_connect_slot(fc_sa, fc_offset, fc_dbname);
			fc_sa->slots[fc_offset].inUse = true;
			return &fc_sa->slots[fc_offset];
		}

		/*
		 * 第四选择：阻塞直到一个或多个槽变得可用。如果
		 * 任何槽遇到致命错误，我们将在这里发现并
		 * 返回NULL。
		 */
		if (!fc_wait_on_slots(fc_sa))
			return NULL;
	}
}

/*
 * ParallelSlotsSetup
 *		准备一组并行槽，但不连接到任何数据库。
 *
 * 这创建并初始化一组槽，将所有并行槽标记为
 * 自由并准备使用。建立连接的过程延迟到请求
 * 一个空闲槽时。cparams、progname、echo和initcmd
 * 被存储以备后用，并且必须在返回数组的生命周期内保持有效。
 */
ParallelSlotArray *
ParallelSlotsSetup(int fc_numslots, ConnParams *fc_cparams, const char *fc_progname,
				   bool fc_echo, const char *fc_initcmd)
{
	ParallelSlotArray *fc_sa;

	Assert(fc_numslots > 0);
	Assert(fc_cparams != NULL);
	Assert(fc_progname != NULL);

	fc_sa = (ParallelSlotArray *) palloc0(offsetof(ParallelSlotArray, slots) +
									   fc_numslots * sizeof(ParallelSlot));

	fc_sa->numslots = fc_numslots;
	fc_sa->cparams = fc_cparams;
	fc_sa->progname = fc_progname;
	fc_sa->echo = fc_echo;
	fc_sa->initcmd = fc_initcmd;

	return fc_sa;
}

/*
 * ParallelSlotsAdoptConn
 *		为重用分配一个打开的连接到插槽数组。
 *
 * 这将将一个打开连接的所有权交给插槽数组。调用者不应进一步使用或关闭该连接。除非可能的 dbname，所有连接的参数（用户、主机、端口等）应与插槽数组的 cparams 匹配，正如在 ParallelSlotsSetup 中所给定的。如果这些参数不同，后续行为是不确定的。
 */
void ParallelSlotsAdoptConn(ParallelSlotArray *fc_sa, PGconn *fc_conn)
{
	int			fc_offset;

	fc_offset = fc_find_unconnected_slot(fc_sa);
	if (fc_offset >= 0)
		fc_sa->slots[fc_offset].connection = fc_conn;
	else
		disconnectDatabase(fc_conn);
}

/*
 * ParallelSlotsTerminate
 *		清理一组并行插槽
 *
 * 迭代给定的 ParallelSlots 中的所有连接，并终止所有连接。
 */
void ParallelSlotsTerminate(ParallelSlotArray *fc_sa)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
	{
		PGconn	   *fc_conn = fc_sa->slots[fc_i].connection;

		if (fc_conn == NULL)
			continue;

		disconnectDatabase(fc_conn);
	}
}

/*
 * ParallelSlotsWaitCompletion
 *
 * 等待所有连接完成，如果在过程中发现至少一个错误，则返回 false。
 */
bool ParallelSlotsWaitCompletion(ParallelSlotArray *fc_sa)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_sa->numslots; fc_i++)
	{
		if (fc_sa->slots[fc_i].connection == NULL)
			continue;
		if (!fc_consumeQueryResult(&fc_sa->slots[fc_i]))
			return false;
	}

	return true;
}

/*
 * TableCommandResultHandler
 *
 * 并行插槽结果处理器，用于命令（而不是查询）对表的结果。
 *
 * 要求结果状态为 PGRES_COMMAND_OK 或关于缺少表的错误。这对于编译要处理的表列表的实用程序非常有用，然后针对这些表运行命令（vacuum、reindex 或其他），因为在列表编译和命令尝试打开表之间存在竞争条件。
 *
 * 对于缺少的表，记录错误但允许处理继续。
 *
 * 对于所有其他错误，记录错误并终止进一步处理。
 *
 * res: 从插槽的连接上执行的查询的 PGresult
 * conn: 属于插槽的连接
 * context: 未使用
 */
bool TableCommandResultHandler(PGresult *fc_res, PGconn *fc_conn, void *fc_context)
{
	Assert(fc_res != NULL);
	Assert(fc_conn != NULL);

	/*
	 * 如果是错误，报告它。缺少表的错误是无害的，所以我们继续处理；但对于其他错误则终止。
	 */
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		char	   *fc_sqlState = PQresultErrorField(fc_res, PG_DIAG_SQLSTATE);

		pg_log_error("processing of database \"%s\" failed: %s",
					 PQdb(fc_conn), PQerrorMessage(fc_conn));

		if (fc_sqlState && strcmp(fc_sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
		{
			PQclear(fc_res);
			return false;
		}
	}

	return true;
}
